package streamforce.crypto.cloud;

import java.io.IOException;
import java.io.ObjectInputStream;
import java.io.ObjectOutputStream;
import java.net.Socket;
import java.util.concurrent.Executor;
import java.util.concurrent.ThreadPoolExecutor;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;

import streamforce.Stat;
import streamforce.crypto.KPAbe;
import streamforce.crypto.cloud.subscriber.CloudSelectPrice;
import streamforce.crypto.cloud.subscriber.CloudSelectAll;
import streamforce.crypto.cloud.subscriber.CloudSumPriceAgg1;
import streamforce.crypto.cloud.subscriber.CloudSumPriceAgg2;
import streamforce.crypto.cloud.subscriber.CloudSumPriceAgg3;
import streamforce.crypto.cloud.subscriber.CloudSumAllAgg1;
import streamforce.crypto.cloud.subscriber.CloudSumAllAgg2;
import streamforce.crypto.cloud.subscriber.CloudSumAllAgg3;
import streamforce.crypto.event.AbstractEvent;
import streamforce.crypto.event.StockEvent;
import streamforce.crypto.event.UserInitAggEvent;
import streamforce.crypto.event.UserInitEvent;
import streamforce.crypto.message.AckMessage;
import streamforce.crypto.message.CloseConnection;
import streamforce.crypto.message.DumpStat;
import streamforce.crypto.message.PolicyMessage;
import streamforce.crypto.policies.Policy;

/**
 * Connection from the data owner
 *
 */
public class EPOwnerConnection extends Thread{
	private static final Log log = LogFactory.getLog(EPOwnerConnection.class);
	private Socket socket; 
	private EPCloud cloudInstance;
	private ThreadPoolExecutor executor;
	Stat nEventReceived; 
	ObjectInputStream ois=null; 
	ObjectOutputStream oos = null;
	KPAbe cryptoHandler; 
	Stat acStat;  
	public EPOwnerConnection(Socket socket, EPCloud cloudInstance, ThreadPoolExecutor executor, KPAbe cryptoHandler){
		this.socket = socket; 
		this.cloudInstance = cloudInstance;
		this.executor = executor;
		this.cryptoHandler = cryptoHandler; 
		this.nEventReceived = new Stat("nEventReceived");
		this.acStat = new Stat("nACInstances");
		try {
			this.ois = new ObjectInputStream(socket.getInputStream());
			this.oos = new ObjectOutputStream(socket.getOutputStream());
		} catch (IOException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}
	}
	
	public void run(){
		try {		
			boolean doneInit = false; 
			do{
				final Object o = ois.readObject();
				if (o instanceof PolicyMessage){ //simple select, type 1 and two
					PolicyMessage pm = (PolicyMessage)o;
					switch (pm.getQueryType()) {
					case PolicyMessage.SIMPLE_QUERY:
						this.initializeQuery_Type1(pm);
						break;
					case PolicyMessage.AGG_QUERY_1:
						this.initializeQuery_Agg1(pm);
						break;
					case PolicyMessage.AGG_QUERY_2:
						this.initializeQuery_Agg2(pm);
						break;
					case PolicyMessage.AGG_QUERY_3:
						this.initializeQuery_Agg3(pm);
						break;
					case PolicyMessage.JOIN_QUERY:
						this.initializeJoinQuery(pm);
						break;
					}								
				}
				else if (o instanceof CloseConnection){
					doneInit = true; 
				}				
				else if (o instanceof StockEvent){ //sending event					
					StockEvent se = (StockEvent)o; 
					nEventReceived.update();									
					EPCloud.getEPServiceProvider().getEPRuntime().sendEvent((StockEvent)o);
					EPCloudServer.tupleTime.put(new Tuple(se.getStockId(),
							se.getTimestamp()), new TimeTuple(System.currentTimeMillis(),0));
				}
			}while (!doneInit);
			if (doneInit){
				log.info("Done policy initialization, connection ended"); 
			}
		} catch (Exception e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		} 
	}
	
	private void initializeQuery_Agg1(PolicyMessage o){
		final PolicyMessage pm = o; 		
		executor.execute(new Runnable(){
			public void run() {
				Policy pol=null;
				if (pm.getSelectType() == PolicyMessage.SELECT_PRICE) {
					pol = new CloudSumPriceAgg1(pm.getUserIds()[0], pm
							.getWindowSize(), cryptoHandler);
				} else if (pm.getSelectType() == PolicyMessage.SELECT_ALL) {
					pol = new CloudSumAllAgg1(pm.getUserIds()[0], pm.getWindowSize(), cryptoHandler);
				}
				
				long start = System.currentTimeMillis(); 
				pol.initACInstance(pm.getStreamIds()[0], pm.getUserIds()[0],
						pm.getPubK()[0], pm.gettK()[0], pm.getDetK()[0]);

				cloudInstance.registerQueryPattern(pm.getQueryExpression(),
						pol);
				acStat.update(); 
				UserInitAggEvent uie = new UserInitAggEvent(pm.getStreamIds()[0], pm
						.getUserIds()[0], pm.getPubK(), pm.gettK(), pm.getDecK(),
						pm.getDetK(), pm.getWindowSize(), pm.getwK());
				EPCloud.getEPServiceProvider().getEPRuntime().sendEvent(uie);
				long end = System.currentTimeMillis(); 
				log.info("init agg query 1 time = "+(end-start)); 
				try {
					oos.writeObject(new AckMessage());
					oos.flush();
				} catch (IOException e) {
					// TODO Auto-generated catch block
					e.printStackTrace();
				}				
			}
		});						 
	}
	
	private void initializeQuery_Agg2(PolicyMessage o){
		final PolicyMessage pm = o; 		
		executor.execute(new Runnable() {
			public void run() {
				Policy pol = null;
				if (pm.getSelectType() == PolicyMessage.SELECT_PRICE) {
					pol = new CloudSumPriceAgg2(pm.getUserIds()[0],cryptoHandler);
				} else if (pm.getSelectType() == PolicyMessage.SELECT_ALL) {
					pol = new CloudSumAllAgg2(pm.getUserIds()[0],cryptoHandler);
				}
				long start = System.currentTimeMillis(); 
				pol.initACInstance(pm.getStreamIds()[0], pm.getUserIds()[0],
						pm.getPubK()[0], pm.gettK()[0], pm.getDetK()[0]);
				acStat.update(); 				
				cloudInstance.registerQueryPattern(pm.getQueryExpression(), pol);					
				long end = System.currentTimeMillis();
				log.info("init agg query 2 time = "+(end-start)); 
				
				UserInitEvent uie = new UserInitEvent(pm.getStreamIds()[0], pm
						.getUserIds()[0], pm.getPubK(), pm.gettK(), pm.getDecK(),
						pm.getDetK());
				EPCloud.getEPServiceProvider().getEPRuntime().sendEvent(uie);				
				try {
					oos.writeObject(new AckMessage());
					oos.flush();
				} catch (IOException e) {
					// TODO Auto-generated catch block
					e.printStackTrace();
				}
			}
		});				
	}
	
	private void initializeQuery_Agg3(PolicyMessage o){
		final PolicyMessage pm = o; 		
		executor.execute(new Runnable() {
			public void run() {
				Policy pol = null;
				if (pm.getSelectType() == PolicyMessage.SELECT_PRICE) {
					pol = new CloudSumPriceAgg3(pm.getUserIds()[0], pm.getOtherUserId(), pm
							.getWindowSize(),cryptoHandler);
				} else if (pm.getSelectType() == PolicyMessage.SELECT_ALL) {
					pol = new CloudSumAllAgg3(pm.getUserIds()[0], pm.getOtherUserId(), pm
							.getWindowSize(),cryptoHandler);
				}
				long start = System.currentTimeMillis(); 
				pol.initACInstance(pm.getStreamIds()[0], pm.getUserIds()[0],
						pm.getPubK()[0], pm.gettK()[0], pm.getDetK()[0]);
				pol.initACInstance(pm.getStreamIds()[0], pm.getOtherUserId(),
						pm.getPubK()[0], pm.gettK()[1], pm.getDetK()[0]);
				
				acStat.update(); 
				
				cloudInstance.registerQueryPattern(pm.getQueryExpression(), pol);
				
				UserInitEvent uie = new UserInitEvent(pm.getStreamIds()[0], pm
						.getUserIds()[0], pm.getPubK(), pm.gettK(), pm.getDecK(),
						pm.getDetK());
				uie.setOtherUserId(pm.getOtherUserId());
				cloudInstance.getEPServiceProvider().getEPRuntime().sendEvent(uie);
				long end = System.currentTimeMillis(); 
				log.info("init agg query 3 time = "+(end-start)); 
				try {
					oos.writeObject(new AckMessage());
					oos.flush();
				} catch (IOException e) {
					// TODO Auto-generated catch block
					e.printStackTrace();
				}
			}
		});
				
		log.info("initialized Query Aggregate 3"); 
	}
	
	private void initializeQuery_Type1(PolicyMessage o){ 
		final PolicyMessage pm = o; 		
		executor.execute(new Runnable(){
			public void run() {
				Policy pol=null;
				long start = System.currentTimeMillis(); 
				if (pm.getSelectType() == PolicyMessage.SELECT_PRICE) {
					pol = new CloudSelectPrice(pm.getUserIds()[0],cryptoHandler);
				} else if (pm.getSelectType() == PolicyMessage.SELECT_ALL) {
					pol = new CloudSelectAll(pm.getUserIds()[0],cryptoHandler);
				}
				
				pol.initACInstance(pm.getStreamIds()[0], pm.getUserIds()[0],
						pm.getPubK()[0], pm.gettK()[0], pm.getDetK()[0]);					
				cloudInstance.registerQueryPattern(pm.getQueryExpression(),
						pol);
				acStat.update(); 
				long end = System.currentTimeMillis(); 	
				log.info("init simple query time = "+(end-start)); 
				UserInitEvent uie = new UserInitEvent(pm.getStreamIds()[0], pm
						.getUserIds()[0], pm.getPubK(), pm.gettK(), pm.getDecK(),
						pm.getDetK());
				EPCloud.getEPServiceProvider().getEPRuntime().sendEvent(uie);				
				try {
					oos.writeObject(new AckMessage());
					oos.flush();
				} catch (IOException e) {
					// TODO Auto-generated catch block
					e.printStackTrace();
				}
			}
		});
						
	}
	
	private void initializeJoinQuery(PolicyMessage o){
		final PolicyMessage pm = o;
		UserInitEvent uie = new UserInitEvent(pm.getStreamIds()[0], pm
				.getUserIds()[0], pm.getPubK(), pm.gettK(), pm.getDecK(),
				pm.getDetK());
		uie.setSelectCondition(pm.getQueryExpression()); 
		EPCloud.getEPServiceProvider().getEPRuntime().sendEvent(uie);
		try {
			oos.writeObject(new AckMessage());
			oos.flush();
		} catch (IOException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}
		//defer initializing ACInstance until later		
	}
}

